package com.raylu.day03basic;

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/13.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class Example8Rebalance {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        //随机分配给下游算子
        env.fromElements(1, 2, 3, 4)
                .shuffle()
                .print("shuffle")
                .setParallelism(2);
        //数据平均分配给下游
        env.fromElements(1, 2, 3, 4)
                .rebalance()
                .print("rebalance")
                .setParallelism(2);
        //依次分配到下游，一般用于上游并行度和下游并行度为倍数关系 1-->n倍
        env.fromElements(1, 2, 3, 4)
                .rescale()
                .print("rescale")
                .setParallelism(3);
//
        //每个并行度都会拿到全量的数据
        env.fromElements(1, 2, 3, 4)
                .broadcast()
                .print("broadcast")
                .setParallelism(2);

        //默认，同broadcast
        env.fromElements(1, 2, 3, 4)
                .print("default")
                .setParallelism(2);
//
        //都分配到其中一个并行度中
        env.fromElements(1, 2, 3, 4)
                .global()
                .print("global")
                .setParallelism(2);
//
        //自定义数据分配策略
        env.fromElements(1, 2, 3, 4).partitionCustom(new Partitioner<Integer>() {
                    @Override
                    public int partition(Integer key, int numPartitions) {
                        return key % 2;
                    }
                }, new KeySelector<Integer, Integer>() {
                    @Override
                    public Integer getKey(Integer value) throws Exception {
                        return value;
                    }
                })
                .print("custom").setParallelism(2);

        env.execute();

    }
}
